package com.bf.electroplating.service.compoent;

import com.bf.electroplating.config.MqttConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MQTT服务器组件
 * 在Spring Boot启动时自动启动MQTT服务器
 */
@Slf4j
@Component
public class MQTTServer implements CommandLineRunner {

    @Autowired
    private MqttConfig mqttConfig;

    private ServerSocket serverSocket;
    private ExecutorService executorService;
    private final AtomicBoolean running = new AtomicBoolean(false);
    
    // 存储连接的客户端
    private final ConcurrentHashMap<String, Socket> connectedClients = new ConcurrentHashMap<>();

    @Override
    public void run(String... args) throws Exception {
        if (!mqttConfig.isEnabled()) {
            log.info("MQTT服务器未启用，跳过启动");
            return;
        }

        log.info("正在启动MQTT服务器，端口: {}", mqttConfig.getPort());
        
        try {
            startMqttServer();
            log.info("MQTT服务器启动成功，监听端口: {}", mqttConfig.getPort());
        } catch (Exception e) {
            log.error("MQTT服务器启动失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 启动MQTT服务器
     */
    private void startMqttServer() throws IOException {
        serverSocket = new ServerSocket(mqttConfig.getPort());
        executorService = Executors.newCachedThreadPool();
        running.set(true);

        // 启动监听线程
        executorService.submit(() -> {
            while (running.get()) {
                try {
                    Socket clientSocket = serverSocket.accept();
                    String clientId = "client_" + System.currentTimeMillis();
                    
                    log.info("新的MQTT连接: {} 来自: {}", clientId, clientSocket.getRemoteSocketAddress());
                    
                    // 为每个客户端创建处理线程
                    executorService.submit(() -> handleClient(clientSocket, clientId));
                    
                } catch (IOException e) {
                    if (running.get()) {
                        log.error("接受客户端连接失败: {}", e.getMessage());
                    }
                }
            }
        });
    }

    /**
     * 处理客户端连接
     */
    private void handleClient(Socket clientSocket, String clientId) {
        try {
            connectedClients.put(clientId, clientSocket);
            log.info("客户端连接: {}, 当前连接数: {}", clientId, connectedClients.size());

            // 发送连接确认
            sendConnectAck(clientSocket);

            // 处理客户端消息
            byte[] buffer = new byte[1024];
            while (running.get() && !clientSocket.isClosed()) {
                int bytesRead = clientSocket.getInputStream().read(buffer);
                if (bytesRead > 0) {
                    byte[] message = new byte[bytesRead];
                    System.arraycopy(buffer, 0, message, 0, bytesRead);
                    handleMqttMessage(clientSocket, clientId, message);
                }
            }

        } catch (IOException e) {
            log.error("处理客户端连接异常: {}", e.getMessage());
        } finally {
            connectedClients.remove(clientId);
            log.info("客户端断开: {}, 当前连接数: {}", clientId, connectedClients.size());
            try {
                clientSocket.close();
            } catch (IOException e) {
                log.error("关闭客户端连接失败: {}", e.getMessage());
            }
        }
    }

    /**
     * 处理MQTT消息
     */
    private void handleMqttMessage(Socket clientSocket, String clientId, byte[] data) {
        if (data.length < 2) {
            log.warn("收到无效的MQTT消息，长度: {}", data.length);
            return;
        }

        // 简单的MQTT消息解析
        byte packetType = (byte) ((data[0] & 0xF0) >> 4);
        
        switch (packetType) {
            case 1: // CONNECT
                log.info("收到MQTT连接请求: {}", clientId);
                sendConnectAck(clientSocket);
                break;
            case 3: // PUBLISH
                handlePublish(clientSocket, clientId, data);
                break;
            case 8: // SUBSCRIBE
                log.info("收到订阅请求: {}", clientId);
                sendSubscribeAck(clientSocket);
                break;
            case 12: // PINGREQ
                log.debug("收到心跳请求: {}", clientId);
                sendPingResp(clientSocket);
                break;
            case 14: // DISCONNECT
                log.info("收到断开连接请求: {}", clientId);
                try {
                    clientSocket.close();
                } catch (IOException e) {
                    log.error("关闭连接失败: {}", e.getMessage());
                }
                break;
            default:
                log.info("收到MQTT消息类型: {}", packetType);
                break;
        }
    }

    /**
     * 处理发布消息
     */
    private void handlePublish(Socket clientSocket, String clientId, byte[] data) {
        try {
            // 简单的主题和消息解析
            if (data.length > 3) {
                int topicLength = data[2] & 0xFF;
                if (data.length > 3 + topicLength) {
                    String topic = new String(data, 3, topicLength, "UTF-8");
                    
                    int messageStart = 3 + topicLength;
                    byte[] payload = new byte[data.length - messageStart];
                    System.arraycopy(data, messageStart, payload, 0, payload.length);
                    
                    // 记录消息
                    logMessage(clientId, topic, payload);
                    
                    // 发送发布确认
                    sendPublishAck(clientSocket);
                }
            }
        } catch (Exception e) {
            log.error("解析发布消息失败: {}", e.getMessage());
        }
    }

    /**
     * 记录接收到的消息
     */
    private void logMessage(String clientId, String topic, byte[] payload) {
        String message = new String(payload);
        log.info("收到MQTT消息 - 客户端: {}, 主题: {}, 内容: {}", clientId, topic, message);
    }

    /**
     * 发送连接确认
     */
    private void sendConnectAck(Socket clientSocket) {
        try {
            byte[] ack = {0x20, 0x02, 0x00, 0x00};
            clientSocket.getOutputStream().write(ack);
            clientSocket.getOutputStream().flush();
        } catch (IOException e) {
            log.error("发送连接确认失败: {}", e.getMessage());
        }
    }

    /**
     * 发送发布确认
     */
    private void sendPublishAck(Socket clientSocket) {
        try {
            byte[] ack = {0x40, 0x02, 0x00, 0x00};
            clientSocket.getOutputStream().write(ack);
            clientSocket.getOutputStream().flush();
        } catch (IOException e) {
            log.error("发送发布确认失败: {}", e.getMessage());
        }
    }

    /**
     * 发送订阅确认
     */
    private void sendSubscribeAck(Socket clientSocket) {
        try {
            byte[] ack = {(byte) 0x90, 0x03, 0x00, 0x01, 0x00};
            clientSocket.getOutputStream().write(ack);
            clientSocket.getOutputStream().flush();
        } catch (IOException e) {
            log.error("发送订阅确认失败: {}", e.getMessage());
        }
    }

    /**
     * 发送心跳响应
     */
    private void sendPingResp(Socket clientSocket) {
        try {
            byte[] resp = {(byte) 0xD0, (byte) 0x00};
            clientSocket.getOutputStream().write(resp);
            clientSocket.getOutputStream().flush();
        } catch (IOException e) {
            log.error("发送心跳响应失败: {}", e.getMessage());
        }
    }

    /**
     * 关闭服务器
     */
    public void shutdown() {
        running.set(false);
        
        if (executorService != null) {
            executorService.shutdown();
        }
        
        if (serverSocket != null && !serverSocket.isClosed()) {
            try {
                serverSocket.close();
            } catch (IOException e) {
                log.error("关闭服务器失败: {}", e.getMessage());
            }
        }
        
        // 关闭所有客户端连接
        connectedClients.values().forEach(clientSocket -> {
            try {
                clientSocket.close();
            } catch (IOException e) {
                log.error("关闭客户端连接失败: {}", e.getMessage());
            }
        });
        connectedClients.clear();
        
        log.info("MQTT服务器已停止");
    }

    /**
     * 获取当前连接数
     */
    public int getConnectedClientsCount() {
        return connectedClients.size();
    }
} 